# -*- coding: UTF-8 -*-
# python 连接kafka 消费数据

import sys
import json
from kafka import KafkaConsumer

class Consumer(object):
    def __init__(self, KafkaServerList=['127.0.0.1:9092'], GroupID='TestGroup001', ClientId="Test", Topics=['Test', ]):
        """
                :param KafkaServerList: kafka服务器IP:PORT 列表
                :param GroupID: 消费者组ID
                :param ClientId: 消费者名称
                :param Topic: 主题

        """
        self._kwargs = {
            "bootstrap_servers": KafkaServerList,
            "client_id": ClientId,
            "group_id": GroupID,
            "enable_auto_commit": False,
            "auto_offset_reset": "latest",
            "key_deserializer": lambda m: json.loads(m),
            "value_deserializer": lambda m: json.loads(m),
        }
        try:
            self._consumer = KafkaConsumer(**self._kwargs)
            self._consumer.subscribe(topics=(Topics))
        except Exception as err:
            print("Consumer init failed, %s" % err)

    def consumeMsg(self):
        try:
            while True:
                data = self._consumer.poll(timeout_ms=5, max_records=100)  # 拉取消息，字典类型
                if data:
                    for key in data:
                        consumerrecord = data.get(key)[0] 
                        if consumerrecord != None:
                            # 消息消费逻辑
                            message = {
                                "Topic": consumerrecord.topic,
                                "Partition": consumerrecord.partition,
                                "Offset": consumerrecord.offset,
                                "Key": consumerrecord.key,
                                "Value": consumerrecord.value
                            }
                            print(message)
                            # 消费逻辑执行完毕后在提交偏移量
                            self._consumer.commit()
                        else:
                            print("%s consumerrecord is None." % key)
        except Exception as err:
            print(err)


def main():
    try:
        c = Consumer(KafkaServerList=['10.10.149.20:6667'], Topics=['js_news'])
        c.consumeMsg()
    except Exception as err:
        print(err)


if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()


